#include "db.h"

DBPacket::DBPacket(Register* reg)
: Packet(reg)
, _state(0)
, _is_simple_query(false)
{
}


DBPacket::DBPacket(uint16 opcode, Register* reg)
: Packet(opcode, reg)
, _state(0)
, _is_simple_query(false)
{
}

DBPacket::DBPacket(const DBPacket &r)
: Packet(r)
, _state(r._state)
, _is_simple_query(r._is_simple_query)
, _query_result(r._query_result)
, _simple_query_result(r._simple_query_result)
{
	memset(_sql, 0, sizeof(_sql));
	strncpy(_sql, r._sql, sizeof(_sql) - 1);
}

DBPacket& DBPacket::operator= (const DBPacket& r)
{
	if(this != &r)
	{
		Packet::operator=(r);

		_state = r._state;
		_is_simple_query = r._is_simple_query;

		memset(_sql, 0, sizeof(_sql));
		strncpy(_sql, r._sql, sizeof(_sql) - 1);

		_query_result = r._query_result;
		_simple_query_result = r._simple_query_result;
	}

	return *this;
}

void DBPacket::SimpleQuery(const char *fmt, ...)
{
	_is_simple_query = true;

	memset(_sql, 0, sizeof(_sql));
	
	va_list ap;
	va_start(ap, fmt);
	vsnprintf(_sql, sizeof(_sql) - 1, fmt, ap);
	va_end(ap);
}

void DBPacket::Query(const char *fmt, ...)
{
	_is_simple_query = false;

	memset(_sql, 0, sizeof(_sql));

	va_list ap;
	va_start(ap, fmt);
	vsnprintf(_sql, sizeof(_sql) - 1, fmt, ap);
	va_end(ap);
}

DBOperator::DBOperator()
: _mysql_sync(false)
, _mysql_async(false)
, connected_timestamp_sync(0)
, connected_timestamp_async(0)
{
}

DBOperator::~DBOperator()
{
}

int DBOperator::_Prepare() 
{
	LOGOUT("--- DBOperator::_Prepare()");

	//std::string _host;
	//uint16	    _port;

	//std::string _user;
	//std::string _password;
	//std::string _database;

	_mysql_sync.set_option(SAFE_NEW mysqlpp::ReconnectOption(true));
	_mysql_sync.set_option(SAFE_NEW mysqlpp::ReadTimeoutOption(3));
	_mysql_sync.set_option(SAFE_NEW mysqlpp::WriteTimeoutOption(3));

	_mysql_async.set_option(SAFE_NEW mysqlpp::ReconnectOption(true));
	_mysql_async.set_option(SAFE_NEW mysqlpp::ReadTimeoutOption(3));
	_mysql_async.set_option(SAFE_NEW mysqlpp::WriteTimeoutOption(3));

	if(!_mysql_async.connect(_database.c_str(), _host.c_str(), _user.c_str(), _password.c_str())) 
	{
		ELOG("_mysql_async.connect failed: %s", _mysql_async.error());
		return -1;
	}
	connected_timestamp_async = GetCurTime();

	return 0;
}

int DBOperator::_Kernel()
{
	if(GetCurTime() - connected_timestamp_async > MAX_KEEP_TIME)
	{
		if(_mysql_async.connected())
		{
			DLOG("DBOperator disconnect _mysql_async");
			_mysql_async.disconnect();
		}
	}

	if(_request_queue.WaitItem(1) == 1)
	{
		return 1;
	}

	if(!_mysql_async.connected())
	{
		if (!_mysql_async.connect(_database.c_str(), _host.c_str(), _user.c_str(), _password.c_str())) 
		{
			ELOG("_mysql_async.connect failed: %s", _mysql_async.error());
			
			return -1;
		}
		connected_timestamp_async = GetCurTime();
	}

	if (!_mysql_async.ping())
	{
		ELOG("mysqlpp::ping() failed[%s]", _mysql_async.query().error());

		_mysql_async.disconnect();
		SleepS(2);
		return -1;
	}

	DBPacket pck;
	if(_request_queue.Dequeue(pck, 1) == 1)
		return 1; // time out

	const char *sql = pck.GetSQL();
	if(pck.IsSimpleQuery())
	{
		mysqlpp::Transaction tran(_mysql_async);
		mysqlpp::SimpleResult res = _mysql_async.query().execute(sql);
		if (!res)
		{
			ELOG("mysqlpp::Query::execute() failed[%s], sql[%s]", _mysql_async.query().error(), sql);
			tran.rollback();
			return -1;
		}
		tran.commit();
		pck.SetSimpleResult(res);
	}
	else
	{
		mysqlpp::Transaction tran(_mysql_async);
		mysqlpp::StoreQueryResult res = _mysql_async.query(sql).store();
		if(!res)
		{
			ELOG("mysql_query() failed[%s], sql[%s]", _mysql_async.query().error(), sql);
			tran.rollback();
			return -1;
		}
		tran.commit();
		pck.SetResult(res);	
	}

	return 0;
}

int DBOperator::_Finish()
{
	_mysql_sync.disconnect();
	_mysql_async.disconnect();
	DLOG("--- DBOperator::_Finish()");
	return 0;
}

void DBOperator::AddRequest(DBPacket& packet)	
{
	ASSERT(!IsStopped());
	_request_queue.Enqueue(packet); 
}

int DBOperator::SimpleQuery(mysqlpp::SimpleResult &result, const char *fmt, ...)
{
	assert(!IsStopped());

	char sql[1024];
	memset(sql, 0, sizeof(sql));
	va_list ap;
	va_start(ap, fmt);
	vsnprintf(sql, sizeof(sql) - 1, fmt, ap);
	va_end(ap);

	if(_mysql_sync.connected())
	{
		_mysql_sync.disconnect();
	}

	if (!_mysql_sync.connect(_database.c_str(), _host.c_str(), _user.c_str(), _password.c_str())) 
	{
		ELOG("_mysql_sync.connect failed: %s", _mysql_sync.error());
		return -1;
	}

	connected_timestamp_sync = GetCurTime();

	if (!_mysql_sync.ping())
	{
		ELOG("mysqlpp::ping() failed[%s]", _mysql_sync.query().error());
		return -1;
	}

	//NLOG("DBOperator executes SQL [%s]", sql);
	mysqlpp::Transaction tran(_mysql_sync);
	result = _mysql_sync.query().execute(sql);
	if (!result)
	{
		ELOG("mysqlpp::Query::execute() failed[%s], sql[%s]", _mysql_sync.query().error(), sql);
		tran.rollback();
		_mysql_sync.disconnect();
		return -1;
	}
	tran.commit();
	_mysql_sync.disconnect();

	return 0;
}

int DBOperator::Query(mysqlpp::StoreQueryResult &result, const char *fmt, ...)
{
	assert(!IsStopped());

	char sql[1024];
	memset(sql, 0, sizeof(sql));
	va_list ap;
	va_start(ap, fmt);
	vsnprintf(sql, sizeof(sql) - 1, fmt, ap);
	va_end(ap);


	if(_mysql_sync.connected())
	{
		_mysql_sync.disconnect();
	}

	if (!_mysql_sync.connect(_database.c_str(), _host.c_str(), _user.c_str(), _password.c_str())) 
	{
		ELOG("_mysql_sync.connect failed: %s", _mysql_sync.error());
		return -1;
	}

	connected_timestamp_sync = GetCurTime();

	if (!_mysql_sync.ping())
	{
		ELOG("mysqlpp::ping() failed[%s]", _mysql_sync.query().error());
		return -1;
	}

	//NLOG("DBOperator executes SQL [%s]", sql);
	mysqlpp::Transaction tran(_mysql_sync);
	result = _mysql_sync.query(sql).store();
	if (!result)
	{
		ELOG("mysql_query() failed[%s], sql[%s]", _mysql_sync.query().error(), sql);
		tran.rollback();
		_mysql_sync.disconnect();
		return -1;
	}
	tran.commit();

	_mysql_sync.disconnect();

	return 0;
}